package com.swak.reactivex.transport;

import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;

import com.swak.Constants;
import com.swak.reactivex.transport.exception.ExchangeException;
import com.swak.reactivex.transport.exception.ExchangeException.ExchangeMethod;
import com.swak.registry.URL;

import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.handler.timeout.IdleStateEvent;

@io.netty.channel.ChannelHandler.Sharable
public class NettyServerHandler extends ChannelDuplexHandler {
	private static final Logger logger = LoggerFactory.getLogger(NettyServerHandler.class);

	private final Map<String, Channel> channels;
	private final ChannelHandler handler;
	private final URL url;

	public NettyServerHandler(URL url, ChannelHandler handler, Map<String, Channel> channels) {
		Assert.isTrue(url != null, "url == null");
		Assert.isTrue(handler != null, "handler == null");
		Assert.isTrue(channels != null, "channels == null");
		this.url = url;
		this.handler = handler;
		this.channels = channels;
	}

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		NettyChannel channel = NettyChannel.getOrAddChannel(url, ctx.channel(), handler);
		if (channel != null) {
			channels.put(ctx.channel().id().toString(), channel);
			channel.setAttribute(Constants.SIDE, Constants.SERVER_SIDE);
		}
		handler.connected(channel);

		if (logger.isInfoEnabled()) {
			logger.info("The connection of " + channel.remoteAddress() + " -> " + channel.localAddress()
					+ " is established.");
		}
	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		NettyChannel channel = NettyChannel.getOrAddChannel(url, ctx.channel(), handler);
		try {
			channels.remove(ctx.channel().id().toString());
			handler.disconnected(channel);
		} finally {
			NettyChannel.removeChannel(ctx.channel());
		}

		if (logger.isInfoEnabled()) {
			logger.info("The connection of " + channel.remoteAddress() + " -> " + channel.localAddress()
					+ " is disconnected.");
		}
	}

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
		NettyChannel channel = NettyChannel.getOrAddChannel(url, ctx.channel(), handler);
		try {
			handler.received(channel, msg);
		} catch (Exception t) {
			Exception error = null;
			if (t instanceof ExchangeException || (t.getCause() != null && t.getCause() instanceof ExchangeException)) {
				error = (Exception) t;
			} else {
				error = new ExchangeException(ExchangeMethod.Read, msg, t);
			}
			throw error;
		}
	}

	@Override
	public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
		super.write(ctx, msg, promise);
		final NettyChannel channel = NettyChannel.getOrAddChannel(url, ctx.channel(), handler);

		// We add listeners to make sure our out bound event is correct.
		// If our out bound event has an error (in most cases the encoder fails),
		// we need to have the request return directly instead of blocking the invoke
		// process.
		promise.addListener(future -> {
			if (future.isSuccess()) {
				// if our future is success, mark the future to sent.
				handler.sent(channel, msg);
				return;
			}

			// 封装异常
			Throwable t = future.cause();
			if (t != null) {
				Throwable error = null;
				if (t instanceof ExchangeException
						|| (t.getCause() != null && t.getCause() instanceof ExchangeException)) {
					error = t;
				} else {
					error = new ExchangeException(ExchangeMethod.Write, msg, t);
				}

				// 统一的异常处理
				handler.caught(channel, error);
			}
		});
	}

	@Override
	public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
		// server will close channel when server don't receive any heartbeat from client
		// util timeout.
		if (evt instanceof IdleStateEvent) {
			NettyChannel channel = NettyChannel.getOrAddChannel(url, ctx.channel(), handler);
			try {
				logger.info("IdleStateEvent triggered, close channel " + channel);
				channel.close();
			} finally {
				NettyChannel.removeChannelIfDisconnected(ctx.channel());
			}
		}
		super.userEventTriggered(ctx, evt);
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		NettyChannel channel = NettyChannel.getOrAddChannel(url, ctx.channel(), handler);
		try {
			handler.caught(channel, cause);
		} finally {
			NettyChannel.removeChannelIfDisconnected(ctx.channel());
		}
	}
}
